package com.rrd.dw.mr.gzip.kafka;

import com.rrd.dw.utils.DateUtil;
import com.ucredit.babel.common.model.BaseModel;
import com.ucredit.babel.common.model.RecordType;
import com.ucredit.babel.common.model.callhistory.CallHistoryAppInfo;
import com.ucredit.babel.common.model.callhistory.CallHistoryCallDetail;
import com.ucredit.babel.common.model.callhistory.CallHistoryDeviceInfo;
import com.ucredit.babel.common.model.callhistory.CallHistoryRecord;
import com.ucredit.babel.common.model.callhistory.CallHistoryUserInfo;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper.Context;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

public class NewPhoneCallRecordParse extends ParseBase {
    private Text val = new Text();
    private Text key = new Text();

    /**
     * 手机通话记录
     *
     * @param model
     * @throws InterruptedException
     * @throws IOException
     * @return
     */
    public boolean parseJob(Context context, BaseModel<?> model, String successful) throws IOException, InterruptedException {

        CallHistoryRecord obj = (CallHistoryRecord) model;
        CallHistoryAppInfo app = obj.getAppInfo() == null ? new CallHistoryAppInfo() : obj.getAppInfo();
        CallHistoryUserInfo user = obj.getUserInfo()  == null ? new CallHistoryUserInfo() : obj.getUserInfo();
        CallHistoryDeviceInfo device = obj.getDeviceInfo()  == null ? new CallHistoryDeviceInfo() : obj.getDeviceInfo();
        List<CallHistoryCallDetail> list = obj.getCalls()  == null ? new ArrayList<>(): obj.getCalls();
        StringBuilder sb = getBaseInfo(obj, successful);
        String timestamstr = obj.getTimestamp() == null ? "" : DateUtil.toDateStrMs(new Date(obj.getTimestamp()));
        // 系统唯一Key
        sb.append(user.getSystemUniqueKey() == null ? "" : user.getSystemUniqueKey()).append(CRT001);
        //app的唯一标识
        sb.append(app.getAppBundle() == null ? "" : app.getAppBundle()).append(CRT001);
        //app版本
        sb.append(app.getAppVersion() == null ? "" : app.getAppVersion()).append(CRT001);
        //是否越狱
        sb.append(device.getJailbroken() == null ? "" : device.getJailbroken()).append(CRT001);
        //分辨率
        sb.append(device.getScreen() == null ? "" : device.getScreen()).append(CRT001);
        sb.append(device.getSimulator() == null ? "" : device.getSimulator()).append(CRT001);
        //网络类型
        sb.append(device.getNetworkType() == null ? "" : device.getNetworkType()).append(CRT001);
        //系统
        sb.append(device.getOs() == null ? "" : device.getOs()).append(CRT001);
        //是否wifi
        sb.append(device.getWifi() == null ? "" : device.getWifi()).append(CRT001);
        //设备唯一码
        sb.append(device.getImei() == null ? "" : device.getImei()).append(CRT001);
        //设备名称
        sb.append(device.getDeviceName() == null ? "" : device.getDeviceName()).append(CRT001);
        //语言
        sb.append(device.getLanguage() == null ? "" : device.getLanguage()).append(CRT001);
        //系统版本
        sb.append(device.getOsVersion() == null ? "" : device.getOsVersion()).append(CRT001);
        //model
        sb.append(device.getModel() == null ? "" : device.getModel()).append(CRT001);
        //运营商
        sb.append(device.getCarrier() == null ? "" : device.getCarrier()).append(CRT001);

        setFileNum(20);
        StringBuilder detailSb = new StringBuilder(500);
        int i = 0;
        for (CallHistoryCallDetail c : list) {
            i++;
            detailSb.delete(0, detailSb.length());
            //序号
            detailSb.append(i).append(CRT001);
            //通话类型
            detailSb.append(c.getType() == null ? "" : c.getType()).append(CRT001);
            //通话联系人
            detailSb.append(c.getName() == null ? "" : c.getName()).append(CRT001);
            //通话号码
            detailSb.append(c.getNumber() == null ? "" : c.getNumber()).append(CRT001);
            //通话号码类型，如
            detailSb.append(c.getNumberType() == null ? "" : c.getNumberType()).append(CRT001);
            //号码标签
            detailSb.append(c.getNumberLabel() == null ? "" : c.getNumberLabel()).append(CRT001);
            //通话发生时间戳，单位毫秒
            detailSb.append(c.getDate() == null ? "" : c.getDate()).append(CRT001);
            //通话时长，单位秒
            detailSb.append(c.getDuration() == null ? "" : c.getDuration()).append(CRT001);
            detailSb.append(timestamstr).append(CRT001);

            key.set(getKey());
            val.set((sb.toString() + detailSb.toString()).replace("\n", " ").replace("\r", " "));
            context.write(key, val);
        }

        if (i == 0) {
            key.set(getKey());
            val.set(sb.toString().replace("\n", " ").replace("\r", " "));
            context.write(key, val);
        }
        return true;
    }

    @Override
    public String getTableName() {
        return "kafka_babel_phone_call_records_inc";
    }

    @Override
    public String getType() {
        return RecordType.PHONE_CALLRECORD.toString();
    }

    @Override
    public String getComment() {
        return "通话记录明细";
    }

}
